package com.simple.datax.seda.core;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.simple.datax.seda.api.ConfigInfoIF;
import com.simple.datax.seda.api.EventHandlerIF;
import com.simple.datax.seda.api.QueueElementIF;
import com.simple.datax.seda.api.SelectSourceIF;
import com.simple.datax.seda.api.StageWrapperIF;
import com.simple.datax.seda.core.event.EntityNestedEvent;
import com.simple.datax.seda.core.event.HeartbeatEvent;
import com.simple.datax.seda.core.event.NullEvent;
import com.simple.datax.seda.core.event.StageInterruptEvent;
import com.simple.datax.seda.core.exception.InitFailedException;
import com.simple.datax.seda.core.exception.UnknownStageEventException;

public class SelectSourceEventHandler implements EventHandlerIF {

	private Log LOG = LogFactory.getLog(SelectSourceEventHandler.class);

	private final Object LOCKER = new Object();

	private SelectSourceIF selectSource;

	private StageWrapperIF nextStage;

	private long sleepMilisecond = 1000;

	/**
	 * 加载数据上限,达到系统退出条件
	 */
	private int threshold = 320;

	/**
	 * 
	 */
	private int currCount = 0;

	public void clear() {
		selectSource.clear();
		currCount = 0;
	}

	public SelectSourceEventHandler() throws InitFailedException {
		this.nextStage = null;
		this.selectSource = null;
	}

	public void initialize() throws InitFailedException {
		if (this.nextStage == null) {
			throw new InitFailedException("nextStage属性是空.");
		}
		if (this.selectSource == null) {
			throw new InitFailedException("selectSource属性是空.");
		}
		LOG.info("SelectSourceEventHandler initialize");
		selectSource.initialize();
		currCount = 0;
	}

	public void setCustomConfigurator(ConfigInfoIF config) {
		selectSource.setCustomConfigurator(config);
	}

	public boolean handleEvent(QueueElementIF event) throws Exception {

		if (event instanceof HeartbeatEvent) {
			// TODO
			LOG.debug("HeartbeatEvent 事件..");
			// 
		} else if (event instanceof EntityNestedEvent) {

			EntityNestedEvent entityEvent = (EntityNestedEvent) event;

			if (entityEvent.getErrorCode() != 0) {
				LOG.error("命令对象有错误.详细信息是: " + entityEvent.getErrorMessage());
				LOG.error(entityEvent.getNestedEntity());
			}
			this.selectSource.deregister(entityEvent.getNestedEntity());

		} else if (event instanceof NullEvent) {
			int count = 0;
			int rtn = SelectSourceIF.SELECT_OK;
			while (true) {
				try {
					rtn = this.reload();
				} catch (Exception e) {
					LOG.error("reload异常:", e);
				}
				LOG.debug("加载数据返回值[" + rtn + "].");
				if (rtn == SelectSourceIF.SELECT_OK) {
					count++;
					LOG.trace("成功加载到一条数据.");
				} else {
					break;
				}
			}
			// 没有加载到数据
			if (count <= 0) {
				if (rtn == SelectSourceIF.SELECT_NONE) {
					LOG.info("系统没有可以加载数据.");
					return false;
				} else if (rtn == SelectSourceIF.SELECT_FREEZE) {
					LOG.info("系统加载达到设置上限[" + this.getThreshold() + "],不再继续加载");
					return false;
				} else if (rtn == SelectSourceIF.SELECT_WOULD_BLOCK) {
					try {
						LOG.trace("源线程暂停 " + sleepMilisecond + " 毫秒.");
						Thread.sleep(sleepMilisecond);
					} catch (RuntimeException e) {
						LOG.error(e);
					}
					return true;
				}
			} else {// 加载到了数据
				LOG.info("系统成功加载[" + count + "]条数据");
				LOG.info("系统累计加载[" + this.currCount + "]条数据");
				return true;
			}
		} else if (event instanceof StageInterruptEvent) {
			// 这个事件最好由线程管理器拦截
			LOG.debug("系统线程中断");
			this.nextStage.putEvent(new StageInterruptEvent());
			return false;
		} else {
			throw new UnknownStageEventException("加载处理器接收到未知命令 ["
					+ event.getClass().getName() + "].");
		}

		return true;
	}

	private int reload() throws Exception {
		LOG.debug("开始加载数据...");
		if (currCount >= this.getThreshold()) {
			LOG.debug("当前加载数量[" + currCount + "]大于加载阀值[" + this.getThreshold()
					+ "]");
			selectSource.freeze();
		}
		Object selectedObj = null;
		ObjectHolder objectHolder = new ObjectHolder();
		int rtn = selectSource.select(objectHolder);

		if (rtn != SelectSourceIF.SELECT_OK) {
			// 如果没有获取到待处理数据,并判断源数据加载器是否还有未注销的数据,有的话,线程暂不中止。
			if (selectSource.remainObject() > 0) {
				// 将加载的对象,打包成Event发给next stage
				LOG.debug("当前加载数据为[" + currCount + "]");
				boolean success = this.nextStage.putEvent(new NullEvent());
				if (!success) {
					// 这是一种极为异常的情况,应该避免发生
					// 如果发生了,selectSource应该保证好数据不丢失.
					LOG.error("加载处理器将数据发送队列时发生错误.");
					this.selectSource.deregister(selectedObj);
				}
				LOG.debug("加载处理器仍然有数据需要处理,暂不加载新数据.");
				return SelectSourceIF.SELECT_WOULD_BLOCK;
			}
			return rtn;
		}

		selectedObj = objectHolder.value;
		synchronized (LOCKER) {
			currCount++;
			LOG.info("加载处理器共有[" + this.currCount + "]条数据");
		}

		// 将加载的对象,打包成Event发给next stage
		boolean success = this.nextStage.putEvent(new EntityNestedEvent(
				selectedObj));
		if (!success) {
			// 这是一种极为异常的情况,应该避免发生
			// 如果发生了,selectSource应该保证好数据不丢失.
			LOG.error("加载处理器将数据发送到队列时发生错误.");
			this.selectSource.deregister(selectedObj);
		}
		LOG.debug("加载处理器成功将一条加载的数据打包成处理命令发送给数据处理器.");
		return SelectSourceIF.SELECT_OK;

	}

	public SelectSourceIF getSelectSource() {
		return selectSource;
	}

	public void setSelectSource(SelectSourceIF selectSource) {
		this.selectSource = selectSource;
	}

	public StageWrapperIF getNextStage() {
		return this.nextStage;
	}

	public void setNextStage(StageWrapperIF reloadStage) {
		this.nextStage = reloadStage;
	}

	public int getThreshold() {
		return threshold;
	}

	public void setThreshold(int threshold) {
		this.threshold = threshold;
	}

	public long getSleepMilisecond() {
		return sleepMilisecond;
	}

	public void setSleepMilisecond(long sleepMilisecond) {
		this.sleepMilisecond = sleepMilisecond;
	}

}
